package com.hooray.elasticsearch;

import java.util.List;
import java.util.Map;

import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.hooray.elasticsearch.exception.ElasticSearchException;

/**
 * 
 * @Desc ElasticsearchClient对外操作的接口，封装ElasticsearchClient的方法调用
 * 
 *       使用需先调用初始化init方法，关闭时需调用destory方法
 *
 * @Author daijunjie
 * @DateTime 2016年4月14日 下午4:45:24
 *
 */
public class ElasticsearchManager {

	private static ElasticsearchClient elasticsearchClient;

	public static ElasticsearchManager INSTANCE = new ElasticsearchManager();

	private static Logger logger = LoggerFactory.getLogger(ElasticsearchManager.class);

	private ElasticsearchManager() {
	}

	/**
	 * 初始化
	 * 
	 * @param clusterName
	 * @param hosts
	 */
	public synchronized void init(String clusterName, String hosts) {
		try {
			if (elasticsearchClient == null) {
				elasticsearchClient = new ElasticsearchClient(clusterName, hosts);
			}
			elasticsearchClient.start();
		} catch (Throwable t) {
			throw new ElasticSearchException(t.getMessage(), t);
		}

	}

	/**
	 * 关闭
	 */
	public void destory() {
		try {
			if (elasticsearchClient != null) {
				elasticsearchClient.stop();
			}
		} catch (Throwable t) {
			throw new ElasticSearchException(t.getMessage(), t);
		}

	}

	/**
	 * 获取elasticsearchClient
	 * 
	 * @return
	 */
	public static ElasticsearchClient getElasticsearchClient() {
		return elasticsearchClient;
	}

	/**
	 * 获取elasticsearch transportClient
	 * 
	 * @return
	 */
	public Client getClient() {
		return elasticsearchClient.getClient();
	}

	/**
	 * 创建索引
	 * 
	 * @param index
	 *            索引
	 * @param type
	 *            类型
	 * @param id
	 *            文档ID，可以为null，当传null时，ES会自动创建
	 * @param source
	 *            json格式文档source
	 * @return 文档创建返回true，文档更新返回false
	 */
	public boolean createIndex(String index, String type, String id, Object source) {
		try {
			IndexResponse response = null;
			if (source instanceof Map) {
				response = elasticsearchClient.createIndex(index, type, id, (Map<?, ?>) source);
			} else if (source instanceof String) {
				response = elasticsearchClient.createIndex(index, type, id, (String) source);
			} else if (source instanceof XContentBuilder) {
				response = elasticsearchClient.createIndex(index, type, id, (XContentBuilder) source);
			} else {
				response = elasticsearchClient.createIndex(index, type, id, source);
			}
			return response.isCreated();
		} catch (Throwable t) {
			throw new ElasticSearchException(t.getMessage(), t);
		}
	}
	
	/**
	 * 创建索引
	 * 
	 * @param index
	 *            索引
	 * @param type
	 *            类型
	 * @param source
	 *            json格式文档source
	 * @return 文档创建返回true，文档更新返回false
	 */
	public String createIndex(String index, String type, Object source) {
		try {
			IndexResponse response = null;
			if (source instanceof Map) {
				response = elasticsearchClient.createIndex(index, type, null, (Map<?, ?>) source);
			} else if (source instanceof String) {
				response = elasticsearchClient.createIndex(index, type, null, (String) source);
			} else if (source instanceof XContentBuilder) {
				response = elasticsearchClient.createIndex(index, type, null, (XContentBuilder) source);
			} else {
				response = elasticsearchClient.createIndex(index, type, null, source);
			}
			return response.getId();
		} catch (Throwable t) {
			throw new ElasticSearchException(t.getMessage(), t);
		}
	}
	
	
	/**
	 * 获取索引
	 * 
	 * @param index
	 * @param type
	 * @param id
	 * @return
	 */
	public Map<String, Object> getIndex(String index, String type, String id) {
		GetResponse response = elasticsearchClient.getIndex(index, type, id);
		return response.getSource();
	}

	/**
	 * 删除索引
	 * 
	 * <p>
	 * 如果传入的索引中，有一个不存在，将导致所有删除失败
	 * </p>
	 * 
	 * @param client
	 * @param indices
	 * @return
	 */
	public boolean deleteIndexs(String... indices) {
		try {
			return elasticsearchClient.deleteIndexs(indices);
		} catch (Throwable t) {
			logger.error(t.getMessage(), t);
			return false;
		}
	}

	/**
	 * 批量添加索引，当插入数据有问题时，会丢弃错误的数据，正确的数据会保存，错误会抛出异常
	 * 
	 * <p>
	 * 批量操作时，最好存储相同的数据格式
	 * </p>
	 * 
	 * 
	 * @param index
	 *            索引
	 * @param type
	 *            类型
	 * @param sourceList
	 *            对象支持Map<String, ?>,JSON
	 *            String,XContentBuilder,JavaBean(get,set method)
	 * @throws ElasticSearchException
	 * 
	 */
	public void bulkAddIndex(String index, String type, List<Object> sourceList) {
		try {
			BulkResponse bulkResponse = elasticsearchClient.bulkAddIndex(index, type, sourceList);
			System.out.println(bulkResponse.hasFailures());
			if (bulkResponse.hasFailures()) {
				throw new ElasticSearchException(bulkResponse.buildFailureMessage());
			}
		} catch (Throwable t) {
			throw new ElasticSearchException(t.getMessage(), t);
		}
	}

	/**
	 * 设置批量操作size
	 * 
	 * @param bulkSize
	 */
	public void setBulkSize(int bulkSize) {
		elasticsearchClient.setBulkSize(bulkSize);
	}

}
